package tbdp.tas

import tbdp.tas.model.StockBlock
import tbdp.tas.util.db.DBUtil
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
  * Created by xbkaishui on 16/9/5.
  * run  topic.out.MinsLine_Block2 ZJ-HZ-BN01-KAFKA-02:9092,ZJ-HZ-BN01-KAFKA-03:9092 10
  */
object BKMinsline {

  /**
    * 加载板块股票信息表找到板块和股票的对应关系
    *
    * @return
    */
  def loadStockBlockInfo(): ArrayBuffer[StockBlock] = {
    val conn = Utils.getMysqlConn()
    val bkTable = Constants.mysql_stock_block
    val bkRLTable = Constants.mysql_stock_block_rl

    val bkRlSql = s"select BCode ,MarketType, SCode  from $bkRLTable"
    val bkRlMap = new mutable.HashMap[String, mutable.Set[String]]()
    val bkRlRes = DBUtil.query(conn, bkRlSql)
    while (bkRlRes.next()) {
      val bCode = bkRlRes.getString(1)
      val sCode = bkRlRes.getString(2) + "_" + bkRlRes.getString(3)
      val values = bkRlMap.getOrElseUpdate(bCode, new mutable.HashSet[String]())
      values.add(sCode)
      bkRlMap.put(bCode, values)
    }

//    println("bk stock rl map " + bkRlMap)

    val bkSql = s"select BCode , BName , MarketType , BlockType, SimpleName ,FixValue ,CalStatus from $bkTable"
    val bkStocks = new ArrayBuffer[StockBlock]()
    val res = DBUtil.query(conn, bkSql)
    while (res.next()) {
      val stockBlock = new StockBlock
      stockBlock.bCode = res.getString(1)
      stockBlock.bName = res.getString(2)
      stockBlock.marketType = 9
      stockBlock.blockType = res.getInt(4)
      stockBlock.simpleName = res.getString(5)
      stockBlock.fixValue = res.getLong(6)
      stockBlock.calStatus = res.getInt(7)
      stockBlock.stocks = bkRlMap.getOrElse(stockBlock.bCode, new mutable.HashSet[String]())
      bkStocks.append(stockBlock)
    }
//    println("bk map " + bkStocks)
    bkStocks
  }


  def main(args: Array[String]) {

    val Array(topic, brokers, interval) = args.take(3)
    val dt = {
      if (args.size > 3) {
        args(3)
      } else {
        Utils.getDate()
      }
    }
    println(topic, brokers, interval, dt)
    val oldVolRateTrainMap = DataLoad.loadVolRateTrain(dt)
    val bkInfos: ArrayBuffer[StockBlock] = loadStockBlockInfo()

    val bkMinslineHandler = new BKMinslineHandler(topic, brokers, interval, dt, bkInfos,oldVolRateTrainMap)
    bkMinslineHandler.handler()
  }
}
